import org.apache.directory.shared.kerberos.codec.apRep.actions.ApRepInit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HbaseCreateTable {
    Configuration conf=null;
    Connection conn=null;
    HBaseAdmin admin =null;
    @Before
    public void init () throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","node01");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    //创建表
    @Test
    public void createTable() throws IOException {
        admin = (HBaseAdmin) conn.getAdmin();
        //创建表描述器
        HTableDescriptor teacher = new HTableDescriptor(TableName.valueOf("relationship"));
        //设置列族描述器
        teacher.addFamily(new HColumnDescriptor("friends"));
        //执⾏创建操作
        admin.createTable(teacher);
        System.out.println("relationship表创建成功！！");
    }

    // 添加数据
    @Test
    public void putData() throws IOException {
        //获取table 对象
        Table table = conn.getTable(TableName.valueOf("relationship"));
        List<Put> puts = new ArrayList<Put>();
        // rowkey  uid1
        Put put = new Put(Bytes.toBytes("uid1"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid5"),Bytes.toBytes("uid5"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid7"),Bytes.toBytes("uid7"));
        puts.add(put);

        // rowkey uid2
        put = new Put(Bytes.toBytes("uid2"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid6"),Bytes.toBytes("uid6"));
        puts.add(put);

        // rowkey uid3
        put = new Put(Bytes.toBytes("uid3"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
        puts.add(put);

        // rowkey uid5
        put = new Put(Bytes.toBytes("uid5"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        puts.add(put);

        // rowkey uid6
        put = new Put(Bytes.toBytes("uid6"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
        puts.add(put);

        // rowkey uid7
        put = new Put(Bytes.toBytes("uid7"));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        puts.add(put);

        System.out.println(puts);

        table.put(puts);


        //关闭table对象
        table.close();
        System.out.println("插⼊成功！！");
    }

    //删除数据
    @Test
    public void deleteData() throws IOException {
        //需要获取⼀个table对象
        Table table = conn.getTable(TableName.valueOf("relationship"));
        //准备delete对象
        final Delete delete = new Delete(Bytes.toBytes("uid1"));
//执⾏table
        table.delete(delete);
        //关闭table对象
        table.close();
        System.out.println("删除数据成功！！");
    }



    /**
     * 全表扫描
     */
    @Test
    public void scanAllData() throws IOException {
        HTable table = (HTable) conn.getTable(TableName.valueOf("relationship"));
        Scan scan = new Scan();
        ResultScanner resultScanner = table.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();//获取改⾏的所有cell对象
            for (Cell cell : cells) {
                //通过cell获取rowkey,cf,column,value
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "----" + cf + "--" + column + "---" + value);
            }
        }
        table.close();
    }

    @After
    public void destroy(){
        if(admin!=null){
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if(conn !=null){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void deleteFriends(String uid,String friend) throws IOException {
        Table table = conn.getTable(TableName.valueOf("relationship"));
        Delete delete = new Delete(Bytes.toBytes(uid));
        delete.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(friend));
        table.delete(delete);
    }

    @Test
    public void deleteF() throws IOException {
        deleteFriends("uid1","uid2");
    }

}